package com.smsc.headend.task.engine.manager;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.SystemClock;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.smsc.headend.common.utils.RedisUtils;
import com.smsc.headend.module.data.dto.TaskDataDTO;
import com.smsc.headend.module.data.enums.OperationGroup;
import com.smsc.headend.module.data.enums.OperationName;
import com.smsc.headend.module.request.MessageTaskDto;
import com.smsc.headend.module.request.MessageTaskRecordDto;
import com.smsc.headend.module.request.entity.LogRequest;
import com.smsc.headend.module.task.dto.*;
import com.smsc.headend.module.task.entity.DeviceTask;
import com.smsc.headend.task.engine.dao.TaskDAO;
import com.smsc.headend.task.engine.service.task.LogRequestService;
import lombok.extern.slf4j.Slf4j;
import org.assertj.core.util.Maps;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.util.*;
import java.util.stream.Collectors;

/**
 * 管理页面操作的任务相关返回的信息
 */
@Component
@Slf4j
public class OdrRequestManager {

    public static final String ODR_REQUEST_TASK_PROCESSING_ATOMIC_QUEUE_S = "odr.request:task:processing:atomic.queue:{%s}";
    public static final String ODR_REQUEST_TASK_COMPLETE_ATOMIC_QUEUE_S = "odr.request:task:complete:atomic.queue:{%s}";
    public static final String ODR_REQUEST_TASK_FAILED_ATOMIC_QUEUE_S = "odr.request:task:fail:atomic.queue:{%s}";
    public static final String ODR_REQUEST_MESSAGE_ID = "odr.request:message.id:";
    public static final String ODR_REQUEST_MESSAGE_ID_TASK = "odr.request:message.id:task:";
    public static final String ODR_REQUEST_MESSAGE_PROCESSING_TASK_QUEUE_S = "odr.request:message:processing:task.queue:{%s}";
    public static final String ODR_REQUEST_MESSAGE_COMPLETE_TASK_QUEUE_S = "odr.request:message:complete:task.queue:{%s}";
    public static final String ODR_REQUEST_MESSAGE_FAIL_TASK_QUEUE_S = "odr.request:message:fail:task.queue:{%s}";

    public static final int WAITING = 0;
    public static final int PROCESSING = 1;
    public static final int COMPLETED = 2;
    public static final int FAILED = 3;
    public static final String ATOMIC_TASK_END = "atomicTaskEnd";
    public static final String ATOMIC_TASK_START = "atomicTaskStart";
    public static final String TASK_END = "taskEnd";

    @Autowired
    RedisUtils redisUtils;
    @Autowired
    TaskDAO taskDAO;
    @Autowired
    KafkaTemplate kafkaTemplate;
    @Autowired
    LogRequestService logRequestService;


    public void addTask(Task task) {
        if (!task.getIsODR()) {
            return;
        }
        String deviceKey = ODR_REQUEST_MESSAGE_ID + task.getMessageId();
        redisUtils.sadd(deviceKey, task.getTaskNo());
        redisUtils.lpush(ODR_REQUEST_MESSAGE_ID_TASK + task.getMessageId(), JSONUtil.toJsonStr(task));
        redisUtils.expire(ODR_REQUEST_MESSAGE_ID_TASK + task.getMessageId(), 60 * 60);
        redisUtils.expire(deviceKey, 60 * 60);
    }

    public void taskStart(Task task) {
        if (!task.getIsODR()) {
            return;
        }
        if (StringUtils.isEmpty(task.getMessageId())) {
            log.error("task have no messageId. odr request will not update task's status: taskNo={}, type={}", task.getTaskNo(), task.getTaskType());
            return;
        }
        String processingTaskRedisKey = String.format(ODR_REQUEST_MESSAGE_PROCESSING_TASK_QUEUE_S, task.getMessageId());
        redisUtils.sadd(processingTaskRedisKey, task.getTaskNo());
        redisUtils.expire(processingTaskRedisKey, 60 * 60);
//        sendRequestMessageTaskDto(task.getMessageId());
    }

    public void taskEnd(Task task, boolean success) {
        if (!task.getIsODR()) {
            return;
        }
        if (StringUtils.isEmpty(task.getMessageId())) {
            log.error("task have no messageId. odr request will not update task's status: taskNo={}, type={}", task.getTaskNo(), task.getTaskType());
            return;
        }
        log.debug("End Task: {}", task.getTaskNo());
        String processingTaskRedisKey = String.format(ODR_REQUEST_MESSAGE_PROCESSING_TASK_QUEUE_S, task.getMessageId());
        String completeTaskRedisKey = String.format(ODR_REQUEST_MESSAGE_COMPLETE_TASK_QUEUE_S, task.getMessageId());
        String failedTaskRedisKey = String.format(ODR_REQUEST_MESSAGE_FAIL_TASK_QUEUE_S, task.getMessageId());
        redisUtils.smove(processingTaskRedisKey, success ? completeTaskRedisKey : failedTaskRedisKey, task.getTaskNo());
        redisUtils.expire(processingTaskRedisKey, 60 * 60);
        redisUtils.expire(success ? completeTaskRedisKey : failedTaskRedisKey, 60 * 60);
        sendRequestMessageTaskDto(task.getMessageId(), TASK_END);
    }

    public void atomicTaskStart(Task task, AtomicTask atomicTask) {
        if (!task.getIsODR()) {
            return;
        }
        String executingTaskAtomicTasksRedisKey = String.format(ODR_REQUEST_TASK_PROCESSING_ATOMIC_QUEUE_S, task.getTaskNo());
        redisUtils.sadd(executingTaskAtomicTasksRedisKey, atomicTask.getAtomicTaskNo());
        redisUtils.expire(executingTaskAtomicTasksRedisKey, 60 * 60);
        sendRequestMessageTaskDto(task.getMessageId(), ATOMIC_TASK_START);
    }

    public void atomicTaskStartFailed(Task task, AtomicTask atomicTask) {
        if (!task.getIsODR()) {
            return;
        }
        String failedTaskAtomicTasksRedisKey = String.format(ODR_REQUEST_TASK_FAILED_ATOMIC_QUEUE_S, task.getTaskNo());
        redisUtils.sadd(failedTaskAtomicTasksRedisKey, atomicTask.getAtomicTaskNo());
        redisUtils.expire(failedTaskAtomicTasksRedisKey, 60 * 60);
    }

    /**
     * @param task
     * @param atomicTaskNo
     * @param atomicTaskCompleted {@link com.smsc.headend.task.engine.service.task.impl.TaskServiceImpl#callbackTask(com.smsc.headend.module.task.dto.Task, AtomicTaskState, int, com.smsc.headend.common.exception.AbstractUDISException)}
     */
    public void atomicTaskEnd(Task task, String atomicTaskNo, int atomicTaskCompleted) {
        if (!task.getIsODR()) {
            return;
        }

        if (atomicTaskCompleted == 0) {
            return;
        }
        String executingTaskAtomicTasksRedisKey = String.format(ODR_REQUEST_TASK_PROCESSING_ATOMIC_QUEUE_S, task.getTaskNo());
        String completeTaskAtomicTasksRedisKey = String.format(ODR_REQUEST_TASK_COMPLETE_ATOMIC_QUEUE_S, task.getTaskNo());
        String failTaskAtomicTasksRedisKey = String.format(ODR_REQUEST_TASK_FAILED_ATOMIC_QUEUE_S, task.getTaskNo());
        redisUtils.smove(executingTaskAtomicTasksRedisKey, atomicTaskCompleted == -1 ? failTaskAtomicTasksRedisKey : completeTaskAtomicTasksRedisKey, atomicTaskNo);
        redisUtils.expire(executingTaskAtomicTasksRedisKey, 60 * 60);
        redisUtils.expire(atomicTaskCompleted == -1 ? failTaskAtomicTasksRedisKey : completeTaskAtomicTasksRedisKey, 60 * 60);
        sendRequestMessageTaskDto(task.getMessageId(), ATOMIC_TASK_END);
    }

    private void sendRequestMessageTaskDto(String messageId, String type) {
        return;
//        Long key = redisUtils.easyRedisLock(String.format("odr.request:message:lock:%s", messageId));
//        if (key == null) {
//            log.info("odr request: refresh status key failed, messageId={}", messageId);
//            return;
//        }
//        MessageTaskDto dto = getMessageDto(messageId);
//        if (dto == null) {
//            log.error("get message state dto null: msgId={}", messageId);
//            return;
//        }
//        if (CollectionUtils.isEmpty(dto.getDeviceTaskStateDtoList())) {
//            log.debug("get message state dto list fail: msgId={}", messageId);
//            return;
//        }
//        String messageJson = JSONUtil.toJsonStr(dto);
//        log.debug("get messageDto: type={},msg={}" , type, messageJson);
//        kafkaTemplate.send(TaskKafkaTopic.REQUEST_MESSAGE_TASK_STATUS, messageJson.getBytes(StandardCharsets.UTF_8));
//
//        redisUtils.unlockEasyRedisLock(String.format("odr.request:message:lock:%s", messageId), key);
    }

    private void saveRequestTvEnd(String messageId) {
        if (StringUtils.isEmpty(messageId)) {
            return;
        }
        logRequestService.lambdaUpdate().eq(LogRequest::getMessageId, messageId).set(LogRequest::getTvEnd, SystemClock.now() / 1000).update();
    }

    private void cleanRequestMessage(MessageTaskDto dto) {
        log.info("cleaning message cache: msgId={}", dto.getMessageId());
        String messageId = dto.getMessageId();
        long expireTime = 60;
        Set<Object> taskNos = redisUtils.smembers(ODR_REQUEST_MESSAGE_ID + messageId);
        /*redisUtils.del(String.format(ODR_REQUEST_MESSAGE_COMPLETE_TASK_QUEUE_S, messageId),
                String.format(ODR_REQUEST_MESSAGE_PROCESSING_TASK_QUEUE_S, messageId),
                ODR_REQUEST_MESSAGE_ID + messageId,
                ODR_REQUEST_MESSAGE_ID_TASK + messageId
        );*/
        redisUtils.expire(String.format(ODR_REQUEST_MESSAGE_COMPLETE_TASK_QUEUE_S, messageId), expireTime);
        redisUtils.expire(String.format(ODR_REQUEST_MESSAGE_PROCESSING_TASK_QUEUE_S, messageId), expireTime);
        redisUtils.expire(ODR_REQUEST_MESSAGE_ID + messageId, expireTime);
        redisUtils.expire(ODR_REQUEST_MESSAGE_ID_TASK + messageId, expireTime);

        for (Object t : taskNos) {
            String taskNo = (String) t;
            redisUtils.expire(String.format(ODR_REQUEST_TASK_PROCESSING_ATOMIC_QUEUE_S, taskNo), expireTime);
            redisUtils.expire(String.format(ODR_REQUEST_TASK_COMPLETE_ATOMIC_QUEUE_S, taskNo), expireTime);
            /*redisUtils.del(String.format(ODR_REQUEST_TASK_PROCESSING_ATOMIC_QUEUE_S, taskNo),
                    String.format(ODR_REQUEST_TASK_COMPLETE_ATOMIC_QUEUE_S, taskNo)
            );*/
        }
    }

    private boolean isAllStatePushComplete(MessageTaskDto dto) {
        if (CollectionUtils.isEmpty(dto.getDeviceTaskStateDtoList())) {
            log.error("device task state list empty: messageId={}", dto.getMessageId());
            return false;
        }
        Boolean isAnyDeviceTaskNotComplete = dto.getDeviceTaskStateDtoList().stream()
                .filter(deviceTaskStateDto -> deviceTaskStateDto.getTaskStateDtoList().stream().filter(taskStateDto -> (taskStateDto.getState() != COMPLETED && taskStateDto.getState() != FAILED)).findAny().isPresent())
                .findAny().isPresent();
        //All task complete
        if (!isAnyDeviceTaskNotComplete) {
            return true;
        }
        return false;
    }

    public MessageTaskDto getMessageDto(String messageId) {
        MessageTaskDto dto = new MessageTaskDto();
        dto.setMessageId(messageId);
        dto.setDeviceTaskStateDtoList(getDeviceTaskStateDto(messageId));
        if (isAllStatePushComplete(dto)) {
            try {
                Thread.sleep(200);
            } catch (InterruptedException e) {
            }
            saveRequestTvEnd(messageId);
            dto.setCompleted(1);
            cleanRequestMessage(dto);
        }
        dto.setRecordDtoList(getMessageTaskRecordDto(messageId));
        log.debug("refresh messageTaskDto: {}", JSONUtil.toJsonStr(dto));
        return dto;
    }

    private List<MessageTaskRecordDto> getMessageTaskRecordDto(String messageId) {
        Set<Object> taskNos = redisUtils.smembers(ODR_REQUEST_MESSAGE_ID + messageId);
        List<MessageTaskRecordDto> resultList = new ArrayList<>();
        for (Object t : taskNos) {
            String taskNo = (String) t;
            List<TaskDataDTO> taskDataDTOS = taskDAO.popAllTaskData(taskNo);
            if (CollectionUtils.isEmpty(taskDataDTOS)) {
                continue;
            }

            List<MessageTaskRecordDto> oneTaskRecordList = taskDataDTOS.stream().map(d -> this.convertTaskDataDtoToMessageTaskRecord(d, taskNo))
                    .collect(Collectors.groupingBy(d -> d.getAtomicTaskNo()))
                    .entrySet().stream()
                    .map(this::atomicTaskResultConcat).map(o -> {
                        o.setTaskNo(taskNo);
                        return o;
                    }).collect(Collectors.toList());

            resultList.addAll(oneTaskRecordList);
        }
        return resultList;
    }

    private MessageTaskRecordDto atomicTaskResultConcat(Map.Entry<String, List<MessageTaskRecordDto>> entry) {
        MessageTaskRecordDto result = new MessageTaskRecordDto();
        Map<String, Object> combinedMap = entry.getValue().stream().map(dto -> dto.getHandlerResult()).reduce((r1, r2) -> {
            r1.putAll(r2);
            return r1;
        }).orElse(null);
        result.setAtomicTaskNo(entry.getKey());
        result.setHandlerResult(combinedMap);
        return result;
    }

    private MessageTaskRecordDto convertTaskDataDtoToMessageTaskRecord(TaskDataDTO taskDataDTO, String taskNo) {
        MessageTaskRecordDto result = new MessageTaskRecordDto();
        result.setTaskNo(taskNo);
        result.setAtomicTaskNo(taskDataDTO.getAtomicTaskNo());
        result.setHandlerResult(Maps.newHashMap(taskDataDTO.getKey(), taskDataDTO.getValue()));
        return result;
    }

    private List<DeviceTaskStateDto> getDeviceTaskStateDto(String messageId) {
        List<Object> taskList = redisUtils.lRange(ODR_REQUEST_MESSAGE_ID_TASK + messageId, 0, -1);
        Map<String, List<Task>> groupedTasks = new HashMap<>();
        for (Object o : taskList) {
            Task t = JSONUtil.toBean(StrUtil.toString(o), Task.class);
            groupedTasks.computeIfAbsent(t.getComId() + ":" + (t.getMeterId() == null ? 0 : t.getMeterId()), k -> new ArrayList<>()).add(t);
        }

        return groupedTasks.entrySet().stream().map(stringListEntry -> {
            DeviceTaskStateDto dto = new DeviceTaskStateDto();
            String[] comMeterArr = stringListEntry.getKey().split(":");
            dto.setComId(Convert.toLong(comMeterArr[0]));
            Long meterId = Convert.toLong(comMeterArr[1]);
            if (meterId != 0) {
                dto.setMeterId(meterId);
            }
            dto.setTaskStateDtoList(getTaskStateDto(stringListEntry.getValue()));
            return dto;
        }).collect(Collectors.toList());
    }

    private List<TaskStateDto> getTaskStateDto(List<Task> taskList) {
        return taskList.stream().map(task -> {
            TaskStateDto dto = new TaskStateDto();
            dto.setTaskNo(task.getTaskNo());
            dto.setParamJson(task.getParamJson());
            dto.setTaskType(task.getTaskType());
            dto.setTaskGroupStateDtoList(task.getSubTaskGroups().stream().map(this::getTaskGroupStateDto).collect(Collectors.toList()));

            String completeTaskRedisKey = String.format(ODR_REQUEST_MESSAGE_COMPLETE_TASK_QUEUE_S, task.getMessageId());
            if (redisUtils.sHasKey(completeTaskRedisKey, task.getTaskNo())) {
                dto.setState(COMPLETED);
                return dto;
            }
            String failedTaskRedisKey = String.format(ODR_REQUEST_MESSAGE_FAIL_TASK_QUEUE_S, task.getMessageId());
            if (redisUtils.sHasKey(failedTaskRedisKey, task.getTaskNo())) {
                dto.setState(FAILED);
                return dto;
            }

            if (dto.getTaskGroupStateDtoList().stream().filter(o -> o.getState() == 1).findAny().isPresent()) {
                dto.setState(PROCESSING);
                return dto;
            }

            dto.setState(WAITING);
            return dto;
        }).collect(Collectors.toList());
    }

    private TaskGroupStateDto getTaskGroupStateDto(SubTaskGroup subTaskGroup) {
        if (subTaskGroup == null || CollectionUtils.isEmpty(subTaskGroup.getSubTasks())) {
            return null;
        }
        TaskGroupStateDto dto = new TaskGroupStateDto();
        dto.setOperationGroup(OperationGroup.forId(Convert.toInt(subTaskGroup.getSubTasks().get(0).getGroupId())).getName());
        log.debug("group atomic tasks: {}", JSONUtil.toJsonStr(subTaskGroup.getSubTasks()));
        dto.setAtomicTaskStateDtoList(subTaskGroup.getSubTasks().stream().map(this::getAtomicTaskStateDto).collect(Collectors.toList()));
        String taskNo = subTaskGroup.getSubTasks().get(0).getTaskNo();
        if (dto.getAtomicTaskStateDtoList().stream().filter(o -> {
            if (o.getState() == 1) {
                log.debug("task state: processing, taskNo={}, atomicTaskNo={}", taskNo, o.getAtomicTaskNo());
            }
            return o.getState() == 1;
        }).findAny().isPresent()) {
            dto.setState(PROCESSING);
            return dto;
        }

        if (dto.getAtomicTaskStateDtoList().stream().filter(o -> o.getState() != 2).findAny().isPresent()) {
            dto.setState(WAITING);
            return dto;
        }

        dto.setState(COMPLETED);
        return dto;
    }

    private AtomicTaskStateDto getAtomicTaskStateDto(AtomicTask atomicTask) {
        AtomicTaskStateDto stateDto = new AtomicTaskStateDto();
        stateDto.setAtomicTaskNo(atomicTask.getAtomicTaskNo());
        stateDto.setParamJson(atomicTask.getAtomicTaskParam());
        stateDto.setOperationName(OperationName.forId(atomicTask.getOperationId()).getName());
        String completeTaskAtomicTasksRedisKey = String.format(ODR_REQUEST_TASK_COMPLETE_ATOMIC_QUEUE_S, atomicTask.getTaskNo());
        String failTaskAtomicTasksRedisKey = String.format(ODR_REQUEST_TASK_FAILED_ATOMIC_QUEUE_S, atomicTask.getTaskNo());
        Boolean isCompleted = redisUtils.sHasKey(completeTaskAtomicTasksRedisKey, atomicTask.getAtomicTaskNo());
        if (isCompleted) {
            stateDto.setState(COMPLETED);
            return stateDto;
        }
        Boolean isFailed = redisUtils.sHasKey(failTaskAtomicTasksRedisKey, atomicTask.getAtomicTaskNo());
        if (isFailed) {
            stateDto.setState(FAILED);
            return stateDto;
        }

        Boolean isProcessing = redisUtils.sHasKey(String.format(ODR_REQUEST_TASK_PROCESSING_ATOMIC_QUEUE_S, atomicTask.getTaskNo()), atomicTask.getAtomicTaskNo());
        if (isProcessing) {
            stateDto.setState(PROCESSING);
            return stateDto;
        }
        stateDto.setState(WAITING);
        return stateDto;
    }

    public List<DeviceTask> getDeviceTask(String messageId) {
        List<Object> taskList = redisUtils.lRange(ODR_REQUEST_MESSAGE_ID_TASK + messageId, 0, -1);
        if (CollectionUtils.isEmpty(taskList)) {
            return null;
        }
        return taskList.stream().map(o -> {
            Task t = JSONUtil.toBean(StrUtil.toString(o), Task.class);
            DeviceTask deviceTask = new DeviceTask();
            deviceTask.setTaskNo(t.getTaskNo());
            deviceTask.setComId(t.getComId());
            deviceTask.setDeviceId(t.getMeterId());
            return deviceTask;
        }).collect(Collectors.toList());
    }
}
